package com.yjd.bi.report.service.impl;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.List;
import java.util.Map;

import org.apache.log4j.Logger;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.BeansException;
import org.springframework.jdbc.core.BeanPropertyRowMapper;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;

import com.yjd.bi.report.enums.DataChangeRecordStatusEnum;
import com.yjd.bi.report.model.BiDataChangeRecordModel;
import com.yjd.bi.report.model.BiDatabaseInfoModel;
import com.yjd.bi.report.model.DataChangeRecordModel;
import com.yjd.bi.report.service.IBiDataChangeRecordService;
import com.yjd.bi.report.service.IBiDatabaseInfoService;
import com.yjd.bi.report.service.IDataChangeRecordTrsService;
import com.yjd.bi.report.util.BiUtil;
import com.yjd.comm.annotation.ServiceRemoteProxy;
import com.yjd.comm.base.concurrent.task.ITaskHandler;
import com.yjd.comm.base.concurrent.task.ITaskItem;
import com.yjd.comm.base.concurrent.task.TaskDisposeUtil;
import com.yjd.comm.base.enums.DbWREnums;
import com.yjd.comm.rpm.RpmServiceKeyEnum;
import com.yjd.comm.util.FrameUtil;

/**
 * <b>description</b>： 业务库变更记录日志迁移,此类不能在方法中加spring来控制事务，具体事务请放在其他类中进行控制<br>
 * <b>time</b>：2016-4-11下午2:28:56 <br>
 * <b>author</b>： ready likun_557@163.com
 */
@Service("dataChangeRecordTrsService")
public class DataChangeRecordTrsService implements IDataChangeRecordTrsService {

	private Logger logger = Logger.getLogger(getClass());
	/**
	 * 数据库信息服务
	 */
	@ServiceRemoteProxy(serviceUrlResouceKey = RpmServiceKeyEnum.RPM_BI_SERVICE_KEY)
	private IBiDatabaseInfoService biDatabaseInfoService;

	/**
	 * 数据库信息服务
	 */
	@ServiceRemoteProxy(serviceUrlResouceKey = RpmServiceKeyEnum.RPM_BI_SERVICE_KEY)
	private IBiDataChangeRecordService biDataChangeRecordService;

	/**
	 * 将业务库数据变更日志已到bi库
	 * 
	 * @param dsm
	 * @throws Exception
	 */
	@SuppressWarnings("unchecked")
	public void moveDcrmFromBusDsToBiDs(final BiDatabaseInfoModel dsm)
			throws Exception {
		if (BiUtil.validata(dsm)) {
			while (true) {
				BiDatabaseInfoModel dcdsm = this.getDataChangeRecordDb(dsm);
				if (dcdsm == null) {
					break;
				}
				List<DataChangeRecordModel> dataChangeRecordModels = this
						.getChangeRecordModels(dcdsm, FrameUtil.newHashMap(
								"status",
								DataChangeRecordStatusEnum.WAIT_DISPOSE
										.getValue(), "database_code", dsm
										.getCode()), 0, 100);
				if (dataChangeRecordModels != null
						&& (!dataChangeRecordModels.isEmpty())) {
					TaskDisposeUtil.dispose(true, 5, dataChangeRecordModels,
							new ITaskHandler<DataChangeRecordModel>() {
								@Override
								public void dispose(
										ITaskItem<DataChangeRecordModel> item) {
									DataChangeRecordModel dataChangeRecordModel = item
											.getItem();
									String itemJson = FrameUtil.json(FrameUtil
											.newHashMap("id", dsm.getId(),
													"url", dsm.getJdbc_url(),
													"description",
													dsm.getDescription()));
									try {
										DataChangeRecordTrsService.this
												.moveDcrmToBiDsAndSetDcrmSuccess(
														dsm,
														dataChangeRecordModel);
										dataChangeRecordModel
												.setStatus(DataChangeRecordStatusEnum.SUCCESS
														.getValue());
										if (DataChangeRecordTrsService.this.logger
												.isInfoEnabled()) {
											DataChangeRecordTrsService.this.logger.info(String
													.format("业务变更日志数据迁移成功[ds:%s]，数据[%s]",
															itemJson,
															FrameUtil
																	.json(dataChangeRecordModel)));
										}
									} catch (Exception e) {
										DataChangeRecordTrsService.this.logger.error("moveDcrmToBiDsAndSetDcrmSuccess 数据出错,BiDatabaseInfoModel:"
												+ itemJson
												+ "; DataChangeRecordModel"
												+ FrameUtil
														.json(dataChangeRecordModel));
										DataChangeRecordTrsService.this.logger
												.error("moveDcrmToBiDsAndSetDcrmSuccess 数据出错"
														+ e.getMessage(), e);
										throw new RuntimeException(e);
									}
								}
							});
				} else {
					break;
				}
			}
			// 将业务库中已处理的变更记录移到日志库
			moveBusDbDcrmToDcrmLogs(dsm);
		}
	}

	/**
	 * 将业务库中已处理的变更记录移到日志库
	 * 
	 * @param dsm
	 * @throws Exception
	 */
	@SuppressWarnings("unchecked")
	public void moveBusDbDcrmToDcrmLogs(BiDatabaseInfoModel dsm)
			throws Exception {
		BiDatabaseInfoModel dcdsm = this.getDataChangeRecordDb(dsm);
		if (dcdsm == null) {
			return;
		}
		List<DataChangeRecordModel> list = this
				.getChangeRecordModels(dcdsm, FrameUtil.newHashMap(
						"status",
						DataChangeRecordStatusEnum.SUCCESS
								.getValue(), "database_code", dsm
								.getCode()), 0, 100);
		// 将最后一条去除，（ps：若数据数据都被删除了，此时数据库若重启了，主键会从1开始）
		if (list != null && list.size() >= 2) {
			list.remove(list.size() - 1);
			this.moveBusDbDcrmToDcrmLogs(dsm, list);
		}
	}

	/**
	 * 将业务库中已处理的变更记录移到日志库
	 * 
	 * @param dsm
	 *            数据库
	 * @param list
	 *            需要转移变更记录列表
	 */
	public void moveBusDbDcrmToDcrmLogs(BiDatabaseInfoModel dsm,
			List<DataChangeRecordModel> list) throws Exception {
		if (list != null && !list.isEmpty()) {
			for (DataChangeRecordModel m : list) {
				moveBusDbDcrmToDcrmLog(dsm, m);
			}
		}
	}

	/**
	 * 将m从dsm库中变更记录表移到变更日志表
	 * 
	 * @param dsm
	 *            数据库
	 * @param m
	 *            需要转移的变更记录
	 */
	public void moveBusDbDcrmToDcrmLog(BiDatabaseInfoModel dsm,
			final DataChangeRecordModel m) throws Exception {
		BiDatabaseInfoModel dcdsm = this.getDataChangeRecordDb(dsm);
		if (dcdsm == null) {
			return;
		}
		if (m != null
				&& DataChangeRecordStatusEnum.SUCCESS.getValue().equals(
						m.getStatus()) && m.getId() != null) {
			// 最后留一条，避免mysql重启之后自增从1开始
			int count = BiUtil.getJdbcTemplate(dcdsm).queryForInt(
					"select count(*) from yjd_data_change_record where id>?",
					m.getId());
			if (count > 0) {
				BiUtil.execute(dcdsm, new BiUtil.ITransExecute<Boolean>() {
					@Override
					public Boolean execute(JdbcTemplate jt) throws Exception {
						jt.update(
								"delete from yjd_data_change_record where id=?",
								m.getId());
						jt.update(
								"INSERT INTO yjd_data_change_record_log "
										+ "(id, database_code, table_name, ref_id, type, create_time, status, actual_dispose_time, version)"
										+ " VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
								m.getId(), m.getDatabase_code(),
								m.getTable_name(), m.getRef_id(), m.getType(),
								m.getCreate_time(), m.getStatus(),
								m.getActual_dispose_time(), m.getVersion());
						return true;
					}
				});
			}
		}
	}

	/**
	 * 将dsm库中的dm记录移到bi库,并更新业务库数据为完成状态
	 * 
	 * @param dsm
	 * @param dm
	 * @throws Exception
	 * @throws BeansException
	 */
	public void moveDcrmToBiDsAndSetDcrmSuccess(BiDatabaseInfoModel dsm,
			DataChangeRecordModel dcrm) throws Exception {
		// 将业务库中数据便跟记录移到bi库
		this.moveDcrmToBiDs(dsm, dcrm);
		// 将业务库中数据记录状态变更为完成状态
		this.updateBiDatabaseInfoModelToSuccess(dsm, dcrm.getId());
	}

	/**
	 * 将dsm库中的dm记录移到bi库
	 * 
	 * @param dsm
	 * @param dm
	 * @throws Exception
	 * @throws BeansException
	 */
	public void moveDcrmToBiDs(BiDatabaseInfoModel dsm,
			DataChangeRecordModel dcrm) throws Exception {
		if (dsm != null
				&& dcrm != null
				&& dsm.getId() != null
				&& dcrm.getId() != null
				&& DataChangeRecordStatusEnum.WAIT_DISPOSE.getValue().equals(
						dcrm.getStatus())
				&& (!this.biDataChangeRecordService.isExists(dsm.getId(),
						dcrm.getId(), DbWREnums.WRITE))) {

			BiDataChangeRecordModel biDataChangeRecordModel = new BiDataChangeRecordModel();
			BeanUtils.copyProperties(dcrm, biDataChangeRecordModel);
			biDataChangeRecordModel.setId(null);
			biDataChangeRecordModel.setDatabase_info_id(dsm.getId());
			biDataChangeRecordModel.setData_change_record_id(dcrm.getId());
			// 将业务库中数据便跟记录移到bi库
			this.biDataChangeRecordService.insert(biDataChangeRecordModel);
		}
	}

	/**
	 * 将业务库中数据记录状态变更为完成状态
	 * 
	 * @throws Exception
	 */
	public int updateBiDatabaseInfoModelToSuccess(BiDatabaseInfoModel dsm,
			long id) throws Exception {
		BiDatabaseInfoModel dcdsm = this.getDataChangeRecordDb(dsm);
		if (dcdsm == null) {
			return 0;
		}
		return BiUtil
				.getJdbcTemplate(dcdsm)
				.update("update yjd_data_change_record set version = version + 1,status = ? where id = ?",
						DataChangeRecordStatusEnum.SUCCESS.getValue(), id);
	}

	/**
	 * 获取业务库中数据变更记录
	 * 
	 * @param dsm
	 * @return
	 */
	@SuppressWarnings("unchecked")
	public List<DataChangeRecordModel> getChangeRecordModels(
			BiDatabaseInfoModel dsm, Map<String, Object> paramMap, int page,
			int pageSize) throws Exception {
		if (dsm == null) {
			return null;
		} else {
			JdbcTemplate jt = BiUtil.getJdbcTemplate(dsm);
			List<Object> sqlPms = FrameUtil.newArrayList();
			List<String> whereList = FrameUtil.newArrayList();
			StringBuilder sqlb = new StringBuilder(
					"select * from yjd_data_change_record");
			Object ordero = null;
			if (paramMap != null) {
				StringBuilder sbWhere = new StringBuilder();
				Object status = paramMap.get("status");
				if (status != null) {
					whereList.add("status = ?");
					sqlPms.add(status);
				}
				Object database_code = paramMap.get("database_code");
				if (database_code != null) {
					whereList.add("database_code = ?");
					sqlPms.add(database_code);
				}

				if (whereList != null && !whereList.isEmpty()) {
					int i = 0;
					for (String string : whereList) {
						if (i >= 1) {
							sbWhere.append(" and ");
						}
						sbWhere.append(" ").append(string).append(" ");
						i++;
					}
				}
				if (sbWhere.length() >= 0) {
					sqlb.append(" where ").append(sbWhere);
				}
				ordero = paramMap.get("order");
			}
			if (ordero != null) {
				sqlb.append(" order by " + ordero);
			} else {
				sqlb.append(" order by id asc");

			}
			sqlb.append(" limit ?,?");
			int skip = (page - 1) * pageSize;
			sqlPms.add(skip < 0 ? 0 : skip);
			sqlPms.add(pageSize);
			List<DataChangeRecordModel> list = jt.query(sqlb.toString(),
					new BeanPropertyRowMapper<DataChangeRecordModel>(
							DataChangeRecordModel.class), sqlPms.toArray());
			return list;
		}
	}

	/**
	 * 获取存放dsm库变更日志的数据库
	 * 
	 * @param dsm
	 * @return
	 * @throws Exception
	 */
	public BiDatabaseInfoModel getDataChangeRecordDb(BiDatabaseInfoModel dsm)
			throws Exception {
		if (dsm != null && dsm.getData_change_log_database_info_id() != null) {
			return this.biDatabaseInfoService.getModelById(
					dsm.getData_change_log_database_info_id(), DbWREnums.READ);
		}
		return null;
	}

	public static void test1(BiDatabaseInfoModel dsm) {
		Connection conn = null;
		PreparedStatement ps = null;
		ResultSet rs = null;
		try {
			Class.forName("com.mysql.jdbc.Driver");
			conn = BiUtil.getDataSource(dsm).getConnection(); /*
															 * DriverManager.
															 * getConnection (
															 * "jdbc:mysql://192.168.11.13:3306/www_yijiedai_com?characterEncoding=UTF-8"
															 * ,
															 * "www_yijiedai_com"
															 * , "yijiedai123" )
															 */
			;
			long startTime = System.currentTimeMillis();
			ps = conn
					.prepareStatement("select count(*) from yjd_data_change_record order by id desc");
			rs = ps.executeQuery();
			while (rs.next()) {
				System.out.println(rs.getInt(1));
			}
			long endTime = System.currentTimeMillis();
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			try {
				if (rs != null) {
					rs.close();
				}
				if (ps != null) {
					ps.close();
				}
				if (conn != null && !conn.isClosed()) {
					conn.close();
				}
			} catch (SQLException e) {
				e.printStackTrace();
			}
		}
	}

}
